package flink;

import bean.student;
import bean.teacher;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple8;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class ConnectTest {
    public static void main(String[] args) throws Exception{
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //获取socket输入流
        DataStreamSource<String> teacher = env.socketTextStream("192.168.1.10", 7000);
        DataStreamSource<String> student = env.socketTextStream("192.168.1.10", 8000);
        //提取watermark
        SingleOutputStreamOperator<teacher> teacherDs = teacher.map(lines -> {
            JSONObject jsonObject = JSONObject.parseObject(lines);
            teacher t = new teacher();
            t.setType(jsonObject.getString("type"));
            t.setName(jsonObject.getString("name"));
            t.setAge(jsonObject.getInteger("age"));
            t.setTime(jsonObject.getString("time"));
            return t;
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<teacher>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<teacher>() {
            @Override
            public long extractTimestamp(teacher element, long recordTimestamp) {
                return Long.valueOf(element.getTime());
            }
        }));

        SingleOutputStreamOperator<student> stdentDs = student.map(lines -> {
            JSONObject jsonObject = JSONObject.parseObject(lines);
            student s = new student();
            s.setType(jsonObject.getString("type"));
            s.setName(jsonObject.getString("name"));
            s.setAge(jsonObject.getInteger("age"));
            s.setTime(jsonObject.getString("time"));
            return s;
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<student>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<student>() {
            @Override
            public long extractTimestamp(student element, long recordTimestamp) {
                return Long.valueOf(element.getTime());
            }
        }));

        //连接两个流
       /* //connect
        teacherDs.connect(stdentDs).map(new CoMapFunction<teacher, student, Tuple4<String,String,Integer,Long>>() {
            @Override
            public Tuple4<String, String, Integer, Long> map1(teacher value) throws Exception {
                return new Tuple4<>(value.getType(),value.getName(),value.getAge(),Long.valueOf(value.getTime()));
            }

            @Override
            public Tuple4<String, String, Integer, Long> map2(student value) throws Exception {
                return  new Tuple4<>(value.getType(),value.getName(),value.getAge(),Long.valueOf(value.getTime()));
            }
        }).print("connect");

        //union
        teacherDs.map(new MapFunction<teacher, Tuple4<String,String,Integer,Long>>() {
            @Override
            public Tuple4<String, String, Integer, Long> map(bean.teacher value) throws Exception {
                return new Tuple4<>(value.getType(),value.getName(),value.getAge(),Long.valueOf(value.getTime()));
            }
        }).union(stdentDs.map(new MapFunction<bean.student, Tuple4<String, String, Integer, Long>>() {
            @Override
            public Tuple4<String, String, Integer, Long> map(bean.student value) throws Exception {
                return new Tuple4<>(value.getType(),value.getName(),value.getAge(),Long.valueOf(value.getTime()));
            }
        })).print("union");*/

/*        teacherDs.print(">>>>>>>>>>>>teacherDs");
        stdentDs.print(">>>>>>>>>>>>stdentDs");*/

        //join
        //inner join
/*        teacherDs.join(stdentDs)
                .where(new KeySelector<bean.teacher, Integer>() {
            @Override
            public Integer getKey(bean.teacher value) throws Exception {
                return value.getAge();
            }
        }).equalTo(new KeySelector<bean.student, Integer>() {
            @Override
            public Integer getKey(bean.student value) throws Exception {
                return value.getAge();
            }
        }).window(TumblingEventTimeWindows.of(Time.minutes(10))).apply(new JoinFunction<bean.teacher, bean.student, Tuple8<String,String,Integer,Long,String,String,Integer,Long>>() {
            @Override
            public Tuple8<String, String, Integer, Long, String, String, Integer, Long> join(bean.teacher first, bean.student second) throws Exception {
                return new Tuple8<>(first.getType(),first.getName(),first.getAge(),Long.valueOf(first.getTime()),second.getType(),second.getName(),second.getAge(),Long.valueOf(second.getTime()));
            }
        }).print("join-TumblingEventTimeWindows");*/

       /* //left join
        teacherDs.coGroup(stdentDs).where(new KeySelector<bean.teacher, Integer>() {
            @Override
            public Integer getKey(bean.teacher value) throws Exception {
                return value.getAge();
            }
        }).equalTo(new KeySelector<bean.student, Integer>() {
            @Override
            public Integer getKey(bean.student value) throws Exception {
                return value.getAge();
            }
        }).window(TumblingEventTimeWindows.of(Time.minutes(10))).apply(new CoGroupFunction<bean.teacher, bean.student, Tuple8<String, String, Integer, Long, String, String, Integer, Long>>() {
            @Override
            public void coGroup(Iterable<bean.teacher> first, Iterable<bean.student> second, Collector<Tuple8<String, String, Integer, Long, String, String, Integer, Long>> out) throws Exception {
                Iterator<bean.teacher> teacherIterator = first.iterator();
                Iterator<bean.student> studentIterator = second.iterator();
                List<teacher> teacherList = new ArrayList<teacher>();
                List<student> studentList = new ArrayList<student>();
                while (teacherIterator.hasNext()){
                    bean.teacher teacherInstance = teacherIterator.next();
                    teacherList.add(teacherInstance);
                }
                while (studentIterator.hasNext()){
                    bean.student studentInstance = studentIterator.next();
                    studentList.add(studentInstance);
                }
                for (int i=0;i<teacherList.size();i++){
                    boolean flag =false;
                    for(int j=0;j<studentList.size();j++){
                            out.collect(new Tuple8<>(teacherList.get(i).getType(),teacherList.get(i).getName(),teacherList.get(i).getAge(),Long.valueOf(teacherList.get(i).getTime()),
                                    studentList.get(j).getType(),studentList.get(j).getName(),studentList.get(j).getAge(),Long.valueOf(studentList.get(j).getTime())));
                            flag=true;
                    }

                    if(!flag){
                        out.collect(new Tuple8<>(teacherList.get(i).getType(),teacherList.get(i).getName(),teacherList.get(i).getAge(),Long.valueOf(teacherList.get(i).getTime()),
                                null,null,null,null));
                    }
                }
            }
        }).print("left-join");*/

        //right join
        teacherDs.coGroup(stdentDs).where(new KeySelector<bean.teacher, Integer>() {
            @Override
            public Integer getKey(bean.teacher value) throws Exception {
                return value.getAge();
            }
        }).equalTo(new KeySelector<bean.student, Integer>() {
            @Override
            public Integer getKey(bean.student value) throws Exception {
                return value.getAge();
            }
        }).window(TumblingEventTimeWindows.of(Time.minutes(10))).apply(new CoGroupFunction<bean.teacher, bean.student, Tuple8<String, String, Integer, Long, String, String, Integer, Long>>() {
            @Override
            public void coGroup(Iterable<bean.teacher> first, Iterable<bean.student> second, Collector<Tuple8<String, String, Integer, Long, String, String, Integer, Long>> out) throws Exception {
                Iterator<bean.teacher> teacherIterator = first.iterator();
                Iterator<bean.student> studentIterator = second.iterator();
                List<teacher> teacherList = new ArrayList<teacher>();
                List<student> studentList = new ArrayList<student>();
                while (teacherIterator.hasNext()){
                    bean.teacher teacherInstance = teacherIterator.next();
                    teacherList.add(teacherInstance);
                }
                while (studentIterator.hasNext()){
                    bean.student studentInstance = studentIterator.next();
                    studentList.add(studentInstance);
                }
                for (int i=0;i<studentList.size();i++){
                    boolean flag =false;
                    for(int j=0;j<teacherList.size();j++){
                        out.collect(new Tuple8<>(studentList.get(i).getType(),studentList.get(i).getName(),studentList.get(i).getAge(),Long.valueOf(studentList.get(i).getTime()),
                                teacherList.get(j).getType(),teacherList.get(j).getName(),teacherList.get(j).getAge(),Long.valueOf(teacherList.get(j).getTime())));
                        flag=true;
                    }

                    if(!flag){
                        out.collect(new Tuple8<>(studentList.get(i).getType(),studentList.get(i).getName(),studentList.get(i).getAge(),Long.valueOf(studentList.get(i).getTime()),
                                null,null,null,null));
                    }
                }
            }
        }).print("right-join");

        //启动程序
        env.execute("ConnectTest");
    }
}
